/*  
 * Copyright 2006-2023 www.anyline.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License. 
 * You may obtain a copy of the License at 
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0 
 * 
 * Unless required by applicable law or agreed to in writing, software 
 * distributed under the License is distributed on an "AS IS" BASIS, 
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 * See the License for the specific language governing permissions and 
 * limitations under the License. 
 * 
 *           
 */ 
 
 
/*
 * Copyright 2006-2023 www.anyline.org
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


package org.anyline.data.elasticsearch.runtime;

import org.anyline.data.adapter.ClientHolder;
import org.anyline.data.elasticsearch.adapter.ElasticSearchClient;
import org.anyline.data.runtime.DataRuntime;
import org.anyline.util.BasicUtil;
import org.anyline.util.BeanUtil;
import org.anyline.util.ConfigTable;
import org.anyline.util.SpringContextUtil;
import org.apache.http.Header;
import org.apache.http.HeaderElement;
import org.apache.http.HeaderElementIterator;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.protocol.HttpClientContext;
import org.apache.http.config.Registry;
import org.apache.http.config.RegistryBuilder;
import org.apache.http.conn.ConnectionKeepAliveStrategy;
import org.apache.http.conn.DnsResolver;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.message.BasicHeader;
import org.apache.http.message.BasicHeaderElementIterator;
import org.apache.http.nio.conn.NoopIOSessionStrategy;
import org.apache.http.nio.conn.SchemeIOSessionStrategy;
import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOReactorException;
import org.apache.http.protocol.HTTP;
import org.apache.http.ssl.SSLContexts;
import org.elasticsearch.client.Node;
import org.elasticsearch.client.RestClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.core.env.Environment;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;

import javax.net.ssl.SSLContext;
import java.util.*;


public class ElasticSearchClientHolder extends ClientHolder {
	private static Logger log = LoggerFactory.getLogger(ElasticSearchClientHolder.class);
	/**
	 * 注册数据源
	 * @param key 数据源名称
	 * @param ds 数据源bean id
	 * @param over 是否允许覆盖已有的数据源
	 * @return RestClient
	 * @throws Exception 异常 Exception
	 */
	private static String addDataSource(String key, String ds, boolean over) throws Exception{
		if(dataSources.contains(key)){
			if(!over){
				throw new Exception("[重复注册][thread:"+Thread.currentThread().getId()+"][key:"+key+"]");
			}else{
				//清空
				ElasticSearchRuntimeHolder.destroy(key);
			}
		}
		if(ConfigTable.IS_DEBUG && log.isInfoEnabled()){
			log.info("[创建elasticsearch数据源][thread:{}][key:{}]", Thread.currentThread().getId(), key);
		}
		reg(key);
		ElasticSearchRuntimeHolder.reg(key);
		return ds;
	}
	private static RestClient addDataSource(String key, RestClient ds, boolean over) throws Exception{
		if(dataSources.contains(key)){
			if(!over){
				throw new Exception("[重复注册][thread:"+Thread.currentThread().getId()+"][key:"+key+"]");
			}else{
				//清空
				ElasticSearchRuntimeHolder.destroy(key);
			}
		}
		if(ConfigTable.IS_DEBUG && log.isInfoEnabled()){
			log.info("[创建elasticsearch数据源][thread:{}][key:{}]", Thread.currentThread().getId(), key);
		}
 		reg(key);
		ElasticSearchRuntimeHolder.reg(key);
 		return ds;
	}

	/**
	 * 注册数据源(生产环境不要调用这个方法，这里只设置几个必需参数用来测试)
	 * @param key 切换数据源依据 默认key=datasource
 	 * @param uri uri
	 * @param database 数据库
	 * @param user 用户名
	 * @param password 密码
	 * @return RestClient
	 * @throws Exception 异常 Exception
	 */
	public static String reg(String key, String uri, String database, String user, String password) throws Exception{
		Map<String, String> param = new HashMap<String, String>();
		param.put("uri", uri);
		param.put("database", database);
		param.put("user", user);
		param.put("password", password);
		return reg(key, param);
	}
	public static String reg(String key, Map param, boolean over) throws Exception{
		return addDataSource(key, reg(key, param), over);
	}

	public static String reg(String key, Map param) throws Exception{
		String ds = build(key, null, param);
		if(null == ds) {//创建数据源失败
			return null;
		}
		return addDataSource(key, ds, true);
	}


	public static RestClient reg(String key, RestClient ds, boolean over) throws Exception{
		return addDataSource(key, ds, over);
	}
	public static RestClient reg(String key, RestClient ds) throws Exception{
		return addDataSource(key, ds, true);
	}
	public static String parseDatabase(String uri){
		//ElasticSearchdb://localhost:27017/mydb
		String database = null;
		if(null != uri && uri.contains(":")){
			String[] tmps = uri.split(":");
			tmps = tmps[tmps.length-1].split("/");
			if(tmps.length>1){
				database = tmps[tmps.length-1];
			}
		}
		return database;
	}
	public static String reg(String key, String prefix, Environment env) {
		try {
			if(BasicUtil.isNotEmpty(prefix) && !prefix.endsWith(".")){
				prefix += ".";
			}
			Map<String,Object> map = new HashMap<String,Object>();
			String type = value(prefix, "type", env, String.class,null);
			if(null == type && !type.endsWith("org.anyline.data.elasticsearch.ds.ElasticSearchDataSource")){
				//只支持type=ElasticSearchDataSource
				return null;
			}
			String client = build(key, prefix, map, env, true);
			if(null == client){//创建数据源失败
				return null;
			}
			addDataSource(key, client, false);
			return client;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}
	/**
	 * 创建数据源
	 * @param key key
	 * @param params 帐号密码等参数
	 * @return bean.id
	 * @throws Exception Exception
	 */
	public static String build(String key, String prefix, Map params, Environment env, boolean over) throws Exception{
		if(dataSources.contains(key)){
			if(!over){
				throw new Exception("[重复注册][thread:"+Thread.currentThread().getId()+"][key:"+key+"]");
			}else{
				//清空
				ElasticSearchRuntimeHolder.destroy(key);
			}
		}
 		String datasource_id = DataRuntime.ANYLINE_DATASOURCE_BEAN_PREFIX + key;
		try {
 			String uris =  (String)BeanUtil.propertyNvl(params,"url","uri");
			if(null == uris && null != prefix && null != env){
				uris = value(prefix, "uri", env, String.class, null);
			}

			String[] hosts = uris.split(",");
			HttpHost[] posts = new HttpHost[hosts.length];
			int idx = 0;
			for(String host:hosts){
				String[] tmps = host.split(":");
				String schema = tmps[0];
				String ip = tmps[1].replace("//","");
				int port = BasicUtil.parseInt(tmps[2], 9200);
				posts[idx++] = new HttpHost(ip, port, schema);
			}
			RestClient rest = RestClient.builder(posts)
					.setRequestConfigCallback(requestConfigBuilder -> {
						//设置连接超时时间
						requestConfigBuilder.setConnectTimeout(value(prefix, "connectTimeout", env, Integer.class,10000));
						requestConfigBuilder.setSocketTimeout(value(prefix, "socketTimeout", env, Integer.class,10000));
						requestConfigBuilder.setConnectionRequestTimeout(value(prefix, "connectionRequestTimeout", env, Integer.class,10000));
						return requestConfigBuilder;
					}).setFailureListener(new RestClient.FailureListener() {
						//某节点失败
						@Override
						public void onFailure(Node node) {
							log.error("[ ElasticSearchClient ] >>  node :{}, host:{},  fail !", node.getName(),
									node.getHost());
						}
					}).setHttpClientConfigCallback(httpSyncClientBuilder -> {
						try {
							//设置信任ssl访问
							SSLContext sslContext = SSLContexts.createDefault();
							// 设置协议http和https对应的处理socket链接工厂的对象
							Registry<SchemeIOSessionStrategy> sessionStrategyRegistry = RegistryBuilder
									.<SchemeIOSessionStrategy>create()
									.register("http", NoopIOSessionStrategy.INSTANCE)
									.register("https", new SSLIOSessionStrategy(sslContext))
									.build();

							// 配置io线程
							IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
									.setIoThreadCount(Runtime.getRuntime().availableProcessors())
									// .setIoThreadCount(2)
									.build();

							ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
							Assert.notNull(ioReactor, "ioReactor init error");

							//使用Httpclient连接池的方式配置(推荐)，同时支持netty，okHttp以及其他http框架
							PoolingNHttpClientConnectionManager poolConnManager = new PoolingNHttpClientConnectionManager(ioReactor,
									null, sessionStrategyRegistry, (DnsResolver) null);
							// 最大连接数
							poolConnManager.setMaxTotal(value(prefix, "maxTotalConnect", env, Integer.class,100));
							// 同路由并发数
							poolConnManager.setDefaultMaxPerRoute(value(prefix, "maxConnectPerRoute", env, Integer.class,10));
							//配置连接池
							httpSyncClientBuilder.setConnectionManager(poolConnManager);
							//设置默认请求头
							List<Header> headers = getDefaultHeaders();
							httpSyncClientBuilder.setDefaultHeaders(headers);
							// 设置长连接策略
							httpSyncClientBuilder.setKeepAliveStrategy(connectionKeepAliveStrategy(null,value(prefix, "keepAliveTime", env, Integer.class,10) ));
							httpSyncClientBuilder.disableAuthCaching();
						} catch (IOReactorException e) {
							log.error("ES的Http异步连接池配置错误", e);
						}
						String user = value(prefix, "user", env, String.class, null);
						String password = value(prefix, "password", env, String.class, null);
						return getHttpAsyncClientBuilder(httpSyncClientBuilder, user, password);
					}).build();
			DefaultListableBeanFactory factory =(DefaultListableBeanFactory) SpringContextUtil.getApplicationContext().getAutowireCapableBeanFactory();
			ElasticSearchClient client = new ElasticSearchClient(rest);
			factory.registerSingleton(datasource_id, client);

		} catch (Exception e) {
			log.error("[注册elasticsearch数据源失败][数据源:{}][msg:{}]", key, e.toString());
			return null;
		}
		return datasource_id;
	}
	public static String build(String key, String prefix, Map params) throws Exception{
		return build(key, prefix, params, null, true);
	}
	/**
	 * 检测数据源是否连接正常
	 * @param ds 数据源名称
	 * @return boolean
	 */
	public static boolean validate(String ds){
		return validate(ElasticSearchRuntimeHolder.getRuntime(ds));
	}
	public static boolean validate(){
		return validate(ElasticSearchRuntimeHolder.getRuntime());
	}
	public static boolean validate(DataRuntime runtime){
		RestClient database = (RestClient) runtime.getProcessor();
		return validate(database);
	}
	public static boolean validate(RestClient client){
		try{
			return client.isRunning();
		}catch (Exception e){
			return false;
		}
	}


	/**
	 * 设置请求头
	 *
	 * @return
	 */
	private static List<Header> getDefaultHeaders() {
		List<Header> headers = new ArrayList<>();
/*        headers.add(new BasicHeader("User-Agent",
                "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/31.0.1650.16 Safari/537" +
                        ".36"));
        headers.add(new BasicHeader("Accept-Encoding", "gzip,deflate"));
        headers.add(new BasicHeader("Accept-Language", "zh-CN"));*/
		headers.add(new BasicHeader("Connection", "Keep-Alive"));
		return headers;
	}
	private static HttpAsyncClientBuilder getHttpAsyncClientBuilder(HttpAsyncClientBuilder httpAsyncClientBuilder, String user, String password) {
		if (ObjectUtils.isEmpty(user) || ObjectUtils.isEmpty(password)) {
			return httpAsyncClientBuilder;
		}
		//账密设置
		CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
		//es账号密码（一般使用,用户elastic）
		credentialsProvider.setCredentials(AuthScope.ANY,
				new UsernamePasswordCredentials(user, password));
		httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
		return httpAsyncClientBuilder;
	}

	/**
	 * 配置长连接保持策略
	 *
	 * @return ConnectionKeepAliveStrategy
	 */
	private static ConnectionKeepAliveStrategy connectionKeepAliveStrategy(Map<String, Integer> keepAliveTargetHost, int keepAliveTime) {
		return (response, context) -> {
			// Honor 'keep-alive' header
			HeaderElementIterator it = new BasicHeaderElementIterator(
					response.headerIterator(HTTP.CONN_KEEP_ALIVE));
			while (it.hasNext()) {
				HeaderElement he = it.nextElement();
				String param = he.getName();
				String value = he.getValue();
				if (value != null && "timeout".equalsIgnoreCase(param)) {
					try {
						return Long.parseLong(value) * 1000;
					} catch (NumberFormatException ignore) {
						log.error("解析长连接过期时间异常", ignore);
					}
				}
			}
			HttpHost target = (HttpHost) context.getAttribute(
					HttpClientContext.HTTP_TARGET_HOST);
			//如果请求目标地址,单独配置了长连接保持时间,使用该配置
			Optional<Map.Entry<String, Integer>> any =
					Optional.ofNullable(keepAliveTargetHost).orElseGet(HashMap::new)
							.entrySet().stream().filter(
									e -> e.getKey().equalsIgnoreCase(target.getHostName())).findAny();
			//否则使用默认长连接保持时间
			return any.map(en -> en.getValue() * 1000L).orElse(keepAliveTime * 1000L);
		};
	}
}
